package com.qf.bootconsumer.listeners;

import com.rabbitmq.client.Channel;
import lombok.extern.java.Log;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.time.LocalDateTime;

@Component
public class RabbitMQConsumer {

    //在消费者上直接控制消息确认模式，如果application.yml中有配置，消费者上也有配置，以消费者的为准
//    @RabbitListener(queues = "boot.que1",ackMode = "AUTO")
    public void consumer1(Message msg) throws UnsupportedEncodingException {

        int i = mockService();

        if (i == 1) {
            String msgBody = new String(msg.getBody(), "utf-8");
            String messageId = msg.getMessageProperties().getMessageId();

            System.out.println("队列1：" + msgBody + "==" + messageId);

            System.out.println("-------------------------");
        }
    }

//    @RabbitListener(queues = "boot.que2")
    public void consumer2(Message msg, Channel channel) throws IOException {

        int i = mockService();

        if (i == 1) {

            String msgBody = new String(msg.getBody(), "utf-8");
            String messageId = msg.getMessageProperties().getMessageId();

            System.out.println("队列2：" + msgBody + "==" + messageId);

            //todo：通过feign调用修改生产端那条消息的状态为 已确认 （通过消息的id来修改）

            //成功提交
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(),true);
        }else{
            //失败，重试
            System.out.println("失败了,重试....");
            //不能无限重试
            num = 1;
            channel.basicNack(msg.getMessageProperties().getDeliveryTag(),true,true);
        }

        System.out.println("-------------------------");
    }

    /*
        假设这个业务处理结果为1，表示成功。业务处理结果为2，表示失败
     */
    int num = 2;
    public int mockService(){
        System.out.println("业务处理....");
        return num;
    }


    /**
     * 演示消息未被消费，进入死信
     */
//    @RabbitListener(queues = "boot.normal.que1")
    public void consumer3(Message msg,Channel channel) throws IOException, InterruptedException {
        System.out.println("消息达到普通队列，但是我不消费....");
        MessageProperties messageProperties = msg.getMessageProperties();

        Thread.sleep(5000);

        //拒绝消息不入队--转到死信队列
        channel.basicReject(messageProperties.getDeliveryTag(),false);
    }

    /**
     *  消费死信队列中的消息，实现消息的延迟消费（延迟消息）
     */
    @RabbitListener(queues = "boot.dead.que")
    public void consumer4(Message msg,Channel channel) throws IOException, InterruptedException {
        System.out.println("检查订单状态，如果未支付，就取消订单：" + LocalDateTime.now());
    }

    /*******************演示幂等问题*********************/
    /*
        ***增：非幂等
            ① 利用redis的setnx解决.可以基于消息的id作为key （使用redis锁机制）
            ② 先查询数据库 --判断是否添加--> 添加操作 （保证这个执行操作必须是原子性的）
                2-1： 加分布式锁
                2-2:  利用数据库的锁
                    select * from 表名 where 列=值 for update  -- 查询时上锁
                    判断
                    insert

                    *** 用这种方式一定要保障上述操作在同一个事务中

            ③ 利用数据库的唯一约束 -- 捕获错误异常，确认提交消息

        删：幂等操作

        改：
            设置型的语句：幂等操作
            ***计算行语句：非幂等
                ① 利用redis的setnx解决.可以基于消息的id作为key （使用redis锁机制）
                ② 状态机,修改时，在表中增加一个状态字段  未修改-->已修改 （使用的是数据库的悲观锁）
                    update 表名 set 列1=列1+值1,列2=列2+值2,...,status='已修改' where id=1 and status='未修改';

        查：幂等操作
     */

    @Autowired
    private StringRedisTemplate redisTemplate;

    @RabbitListener(queues = "boot.mideng.que")
    public void consumer5(Message msg,Channel channel) throws IOException {
        MessageProperties messageProperties = msg.getMessageProperties();
        String messageId = messageProperties.getMessageId();

        Boolean flag = redisTemplate.opsForValue().setIfAbsent(messageId, "1");

        //幂等性保障
        if (flag) {
            try {
                String msgBody = new String(msg.getBody(), "utf-8");
                System.out.println(msgBody + "==" + messageId);
            }catch (Exception e){
                e.printStackTrace();
                //失败
                redisTemplate.delete(messageId);
                channel.basicNack(msg.getMessageProperties().getDeliveryTag(), true,true);
                return;
            }
        }

        channel.basicAck(msg.getMessageProperties().getDeliveryTag(), true);
    }
}
